package com.lytoyo.neihanduanzi.listener;

import cn.hutool.extra.pinyin.PinyinUtil;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.lytoyo.neihanduanzi.mapper.StatusMapper;
import com.lytoyo.neihanduanzi.mapper.UserAttentionMapper;
import com.lytoyo.neihanduanzi.model.blog.BlogInfo;
import com.lytoyo.neihanduanzi.model.dto.BlogInfoDto;
import com.lytoyo.neihanduanzi.model.dto.StatusDto;
import com.lytoyo.neihanduanzi.model.index.BlogIndex;
import com.lytoyo.neihanduanzi.model.index.SuggestIndex;
import com.lytoyo.neihanduanzi.model.index.UserIndex;
import com.lytoyo.neihanduanzi.model.user.ChatMsg;
import com.lytoyo.neihanduanzi.model.user.UserAttention;
import com.lytoyo.neihanduanzi.model.user.UserInfo;
import com.lytoyo.neihanduanzi.repository.BlogIndexRepository;
import com.lytoyo.neihanduanzi.repository.SuggestIndexRepository;
import com.lytoyo.neihanduanzi.repository.UserIndexRepository;
import com.lytoyo.neihanduanzi.service.*;
import com.lytoyo.neihanduanzi.utils.AuthContextHolder;
import com.lytoyo.neihanduanzi.utils.Constant;
import com.lytoyo.neihanduanzi.utils.WebsocketLocalContainerUtil;
import com.rabbitmq.client.Channel;
import io.minio.GetObjectArgs;
import io.minio.MinioClient;
import io.minio.RemoveObjectArgs;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.Exchange;
import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.amqp.rabbit.annotation.QueueBinding;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.elasticsearch.core.suggest.Completion;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;

import java.io.*;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.StandardCopyOption;
import java.util.*;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

/**
 * Package:com.lytoyo.neihanduanzi.listener
 *
 * @ClassName:MessageConsumer
 * @Create:2025/1/14 14:19
 **/
@Component
@Slf4j
public class MessageConsumer {
    @Resource
    private StatusMapper statusMapper;

    @Resource
    private MinioClient minioClient;

    @Value("${minio.bucket}")
    private String bucket;

    @Resource
    private BlogService blogService;

    @Resource
    private ChatMsgService chatMsgService;

    @Resource
    private UserService userService;

    @Resource
    private UserIndexRepository userIndexRepository;

    @Resource
    private CosService cosService;

    @Resource
    private BlogIndexRepository blogIndexRepository;

    @Resource
    private SuggestIndexRepository suggestIndexRepository;

    @Resource
    private RabbitTemplate rabbitTemplate;

    @Resource
    private ThreadPoolExecutor threadPoolExecutor;

    @Resource
    private RedisTemplate redisTemplate;


    @Resource
    private UserAttentionMapper userAttentionMapper;

    /**
     * 初始用户状态信息
     *
     * @param dataString
     * @param message
     * @param channel
     */
    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(value = Constant.QUEUE_NEIHANDUANZI_STATUS, durable = "true"),
            exchange = @Exchange(value = Constant.EXCHANGE_DIRECT_NEIHANDUANZI_STATUS),
            key = {Constant.ROUTING_KEY_NEIHANDUANZI_STATUS}))
    @Transactional(rollbackFor = Exception.class)
    public void processMessage(Long dataString, Message message, Channel channel) {
        long userId = dataString;
        StatusDto statusDto1 = new StatusDto()
                .setUserId(userId)
                .setAmount(0l)
                .setType("0701");
        statusMapper.insert(statusDto1);
        StatusDto statusDto2 = new StatusDto()
                .setUserId(userId)
                .setAmount(0l)
                .setType("0702");
        statusMapper.insert(statusDto2);
        StatusDto statusDto3 = new StatusDto()
                .setUserId(userId)
                .setAmount(0l)
                .setType("0703");
        statusMapper.insert(statusDto3);

        //存入elastic
        UserInfo userInfo = userService.getById(userId);
        UserIndex userIndex = new UserIndex();
        BeanUtils.copyProperties(userInfo, userIndex, UserIndex.class);
        userIndex.setFans(0l);
        //上传
        userIndexRepository.save(userIndex);

    }

    /**
     * 合成分片并删除分片，存储文件与用户关联，上传搜素视频
     *
     * @param map
     * @param message
     * @param channel
     */
    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(value = Constant.QUEUE_NEIHANDUANZI_VIDEO_PART, durable = "true"),
            exchange = @Exchange(value = Constant.EXCHANGE_DIRECT_NEIHANDUANZI_VIDEO_PART),
            key = {Constant.ROUTING_KEY_NEIHANDUANZI_VIDEO_PART}))
    public void deleteVideoPartAndSetUserInfo(Map<String, Object> map, Message message, Channel channel) {

        int chuckCount = (Integer) map.get("chuckCount");
        String uuid = (String) map.get("uuid");
        String fileSuffix = (String) map.get("fileSuffix");
        Integer userId = (Integer) map.get("userId");
        String title = (String) map.get("title");
        InputStream videoStream = null;
        try {
            //删除分片
            for (int i = 0; i < chuckCount; i++) {
                String temName = uuid + i + "." + fileSuffix;
                minioClient.removeObject(RemoveObjectArgs.builder()
                        .bucket(bucket)
                        .object(temName)
                        .build());
            }
        } catch (Exception e) {
            log.info(e.getMessage());
        }
        BlogInfoDto blogInfoDto = new BlogInfoDto();
        InputStream stream = null;
        try {
            String imageUrl = "https://neihanduanzi-1325026956.cos.ap-guangzhou.myqcloud.com/OIP-C.jpg";
            //保存博客到数据库
            if (userId != null) {
                BlogInfo blogInfo = new BlogInfo();
                blogInfo.setUserId(Long.valueOf(userId))
                        .setTitle(title)
                        .setFileName(uuid + "." + fileSuffix)
                        .setBlogType(1)
                        .setLiked(0L)
                        .setTimeOfPlay(0L)
                        .setComment(0L)
                        .setCollect(0L)
                        .setPicture(imageUrl);
                //将视频信息与上传文件的用户关联
                blogService.save(blogInfo);
                BeanUtils.copyProperties(blogInfo,blogInfoDto, BlogInfoDto.class);
            }

            //上传博客信息到搜素引擎
//            UserInfo userInfo = userService.getById(userId);
//            BlogIndex blogIndex = new BlogIndex();
//            BeanUtils.copyProperties(blogInfoDto, blogIndex, BlogIndex.class);
//            blogIndex.setDate(new Date())
//                    .setHeadPhoto(userInfo.getHeadPhoto())
//                    .setNickname(userInfo.getNickname());
//            blogIndexRepository.save(blogIndex);
//            System.out.println("33333");
//
//            //上传博客信息到es
//            // 使用正则表达式根据常见标点符号拆分标题
//            String[] sentences = blogInfoDto.getTitle().split("[？！，。:]");
//            sentences[0] = sentences[0].trim();
//            //补全关键词
//            SuggestIndex suggestIndex = new SuggestIndex();
//            suggestIndex.setId(UUID.randomUUID().toString().replaceAll("-", ""))
//                    .setTitle(sentences[0])
//                    .setKeyword(new Completion(new String[]{sentences[0]}))
//                    .setKeywordPinyin(new Completion(new String[]{PinyinUtil.getPinyin(sentences[0])}))
//                    .setKeywordSequence(new Completion(new String[]{PinyinUtil.getFirstLetter(sentences[0], "")}));
//            suggestIndexRepository.save(suggestIndex);


            //将博客信息推送给粉丝
            //当前在线用户
            Set<Long> onlineUserIds = WebsocketLocalContainerUtil.getAllUserId();
            //查询该用户的所有粉丝
            List<Long> userAttentionsUserIds = this.userAttentionMapper.selectObjs(new LambdaQueryWrapper<UserAttention>()
                    .eq(UserAttention::getUserId, userId)
                    .select(UserAttention::getFan));
            Set<Long> onlineUserFanIds = userAttentionsUserIds.stream().collect(Collectors.toSet());
            Set<Long> offLineUserFanIds = new HashSet<>(userAttentionsUserIds);
            //当前离线用户
            offLineUserFanIds.removeAll(onlineUserIds);
            //当前在线的粉丝
            onlineUserFanIds.removeAll(offLineUserFanIds);

            List<Object> list = new ArrayList<>();
            UserInfo userInfoFan = userService.getById(userId);
            blogInfoDto.setNickname(userInfoFan.getNickname()).setHeadPhoto(userInfoFan.getHeadPhoto());
            list.add(blogInfoDto);
            //为在线用户推送消息
            for (Long id : onlineUserFanIds) {
                WebsocketLocalContainerUtil.sendFollowMsg(id, list);
            }
            //为离线用户存储消息
            for (Long id : offLineUserFanIds) {
                redisTemplate.opsForList().leftPush(Constant.USER_FOLLOW_BLOG_ID_KEY + id, blogInfoDto.getId());
                redisTemplate.expire(Constant.USER_FOLLOW_BLOG_ID_KEY + id, Constant.USER_FOLLOW_BLOG_ID_TTL, TimeUnit.DAYS);
            }



        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            if (null != stream) {
                try {
                    stream.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }
    }

    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(value = "dlx.queue", durable = "true"),
            exchange = @Exchange(value = "dlx.exchange"),
            key = "dlx.routing"
    ))
    public void handleDeadLetterMessage(ChatMsg msg, Message message, Channel channel) {
        Integer retryCount = message.getMessageProperties()
                .getHeader("X-Retry-Count");
        if (retryCount == null) retryCount = 0;
        if (retryCount < 3) {
            // 重试发送
            Integer finalRetryCount = retryCount;
            rabbitTemplate.convertAndSend(
                    "user.message.exchange",
                    "user.message.routing",
                    msg,
                    m -> {
                        m.getMessageProperties().setHeader("X-Retry-Count", finalRetryCount + 1);
                        return m;
                    }
            );
        } else {
            // 持久化到数据库
            chatMsgService.save(msg);
        }
    }

    /**
     * 抽取视频第一帧
     *
     * @param videoPath
     * @return
     * @throws Exception
     */
    /**
     * 抽取视频第一帧（安全增强版）
     */
    public Path extractFirstFrame(Path videoPath) throws IOException, InterruptedException {
        // 1. 准备输出路径（确保目录存在）
        Path outputPath = Files.createTempFile("frame-", ".jpg");

        // 2. 构建FFmpeg命令（处理路径转义）
        String[] command = new String[] {
                "ffmpeg",
                "-y",                    // 覆盖输出文件
                "-i", videoPath.toString(),
                "-vframes", "1",         // 只取1帧
                "-q:v", "2",            // 输出质量（1-31，值越小质量越高）
                "-f", "image2",         // 强制输出格式
                outputPath.toString()
        };

        // 3. 执行命令（捕获完整错误流）
        Process process = new ProcessBuilder(command)
                .redirectErrorStream(true)  // 合并错误流到标准输出
                .start();

        // 4. 读取输出日志（诊断用）
        try (BufferedReader reader = new BufferedReader(
                new InputStreamReader(process.getInputStream()))) {
            String line;
            while ((line = reader.readLine()) != null) {
                log.debug("FFmpeg输出: {}", line);
            }
        }

        // 5. 检查执行结果
        int exitCode = process.waitFor();
        if (exitCode != 0) {
            Files.deleteIfExists(outputPath);  // 清理失败产物
            throw new IOException(String.format(
                    "FFmpeg处理失败，退出码：%d。命令：%s",
                    exitCode, String.join(" ", command)
            ));
        }

        return outputPath;
    }


}
